package com.bigdata

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1_WordCount {

  def main(args: Array[String]): Unit = {

    // 创建spark配置对象
    val conf = new SparkConf()

      /**
       * 指定指定模式为local本地模式
       * 本地模式是不依赖各种开发环境
       * 1 指的是设置分区数为1，通用的设置
       */
      .setMaster("local[2]")
      .setAppName("WordCount") // 设置应用名称

    // 需要一个SparkContext对象
    val sc = new SparkContext(conf)
    // 设置日志的打印级别
    sc.setLogLevel("error")

    // 读取数据
    // 默认mapTask的数量和block（128M）一致的
    // 处理一条数据需要循环500次，
    val linesRDD: RDD[String] = sc.textFile("data/words.txt")

    // 获取分区数, 就相当于并行度
    println("getNumPartitions----" + linesRDD.getNumPartitions)

    /**
     * spark 中提供了很多算子，比scala的集合函数还要丰富
     * 底层是不一样的，spark是一个分布式的计算引擎
     * spark 中RDD对象.函数，把这些函数叫做算子
     * RDD中传递的代码逻辑，是作用在每一个partition上的
     * RDD是不存储数据的
     */


    /**
     * Transformations 算子：不会产生计算的算子
     * map(groupBy(flatMap("java,spark,hadoop".split(",")))) =>(word,count)
     * 在每一个算子上实际上保存的是 计算逻辑，并没有保存数据
     * RDD之间是有依赖的关联关系，血统关系，类似父子关系
     *
     * action算子：
     * foreach算子中调用runJob方法，触发job的启动，能够触发job的算子就叫action算子
     */
    val wordRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    val groupRDD: RDD[(String, Iterable[String])] = wordRDD.groupBy(word => word)

    val result: RDD[(String, Int)] = groupRDD.map(tup => {
      val word: String = tup._1
      val count: Int = tup._2.toList.size
      (word, count)
    })

    // 应用程序中要有Action 类算子来触发 Transformation 类算子执行。
    // 在控制台打印
    result.foreach(println)

    println("*" * 100)

    // 想要使用第二个字段进行排序，对key 进行升序或者降序排序
    val sortRDD: RDD[(String, Int)] = result.sortBy(kv => -kv._2)
    sortRDD.foreach(println)
    // Thread.sleep(Integer.MAX_VALUE)
    // path指的是一个目录
    sortRDD.saveAsTextFile("data/wc_output")


  }

}
